package com.lot.ws.dao;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.QualifierFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import com.lot.ws.dto.AnalyticAvgInput;
import com.lot.ws.dto.ReadingDetails;
import com.lot.ws.resource.MSRResource;
import com.lot.ws.util.DateUtil;

public class MSRDao {

	private static final Logger logger = Logger.getLogger(MSRDao.class);
	private static Configuration conf = null;
	private static final String TABLE_RAW_DATA = "msr-lot-raw";
	private static final String TABLE_PROCESSED_DATA = "msr-lot-processed";
	/**
	 * Initialization
	 */
	static {
		logger.info("Connecting to HBASE...............................");
		conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum",
				"mapr-m3-01.cs.ucl.ac.uk,mapr-m3-02.cs.ucl.ac.uk,mapr-m3-03.cs.ucl.ac.uk");
		// // conf.set("hbase.zookeeper.quorum", "localhost");
		conf.set("hbase.zookeeper.property.clientPort", "5181");
		try {
			HBaseAdmin.checkHBaseAvailable(conf);
		} catch (MasterNotRunningException e) {
			logger.error("MasterNotRunningException...", e);
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			logger.error("ZooKeeperConnectionException...");
			e.printStackTrace();
		}

		logger.debug("HBase is running!");
	}

	public static void addRecord(String rowKey, String family,
			String qualifier, String value) throws Exception {
		try {
			HTable table = new HTable(conf, TABLE_RAW_DATA);
			Put put = new Put(Bytes.toBytes(rowKey));
			put.add(Bytes.toBytes(family), Bytes.toBytes(qualifier),
					Bytes.toBytes(value));
			table.put(put);
			logger.debug("insert recored " + rowKey + " to table "
					+ TABLE_RAW_DATA + " ok.");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static Collection<Double> getReadingsAsSet(String rowKey,
			AnalyticAvgInput input) {

		String dateFromStr = String.valueOf(input.getFrom().getTime());
		dateFromStr = DateUtil.makeFixedLength(dateFromStr);
		logger.debug("The date From in miliseconds is : " + dateFromStr);

		String dateToStr = String.valueOf(input.getTo().getTime());
		dateToStr = DateUtil.makeFixedLength(dateToStr);
		logger.debug("The date To in miliseconds is : " + dateToStr);

		Set<Double> avgs = new TreeSet<Double>();
		try {
			logger.debug("rowKey :::" + rowKey + " input.getTimeScale():::"
					+ input.getTimeScale());

			FilterList filterList = applyFilters(rowKey, input, dateFromStr,
					dateToStr);

			// HTable table = new HTable(HBaseMsrAPI.getConf(),
			// "lotAnalyticsTest2");
			HTable table = new HTable(conf, TABLE_PROCESSED_DATA);
			// table.setAutoFlush(true);
			Scan scan = new Scan();
			scan.setFilter(filterList);
			ResultScanner scanner = table.getScanner(scan);
			for (Result result : scanner) {
				for (KeyValue kv : result.raw()) {
					logger.info(new String(kv.getRow()) + " ");
					logger.debug(new String(kv.getFamily()) + ":");
					logger.debug(new String(kv.getQualifier()) + " ");
					// System.out.print(kv.getTimestamp() + " ");
					logger.debug(new String(kv.getValue()));
					String s = new String(kv.getValue());
					Double avgDouble = Double.valueOf(s);
					avgs.add(avgDouble);
				}
			}

			// scanner.close();
			// table.close();

		} catch (IOException e) {
			e.printStackTrace();
		}

		return avgs;
	}

	public static Map<String, Double> getReadingAsMapWithTime(String rowKey,
			AnalyticAvgInput input) {

		String dateFromStr = String.valueOf(input.getFrom().getTime());
		dateFromStr = DateUtil.makeFixedLength(dateFromStr);
		logger.info("The date From in miliseconds is : " + dateFromStr);

		String dateToStr = String.valueOf(input.getTo().getTime());
		dateToStr = DateUtil.makeFixedLength(dateToStr);
		logger.info("The date To in miliseconds is : " + dateToStr);
		System.out.println("rowKey :" + rowKey);

		Map<String, Double> avgs = new TreeMap<String, Double>();
		try {
			logger.debug("rowKey :::" + rowKey + " input.getTimeScale():::"
					+ input.getTimeScale());

			FilterList filterList = applyFilters(rowKey, input, dateFromStr,
					dateToStr);

			// HTable table = new HTable(HBaseMsrAPI.getConf(),
			// "lotAnalyticsTest2");
			HTable table = new HTable(conf, TABLE_PROCESSED_DATA);
			// table.setAutoFlush(true);
			Scan scan = new Scan();
			scan.setFilter(filterList);
			ResultScanner scanner = table.getScanner(scan);
			for (Result result : scanner) {
				for (KeyValue kv : result.raw()) {
					System.out.print(new String(kv.getRow()) + " ");
					System.out.print(new String(kv.getFamily()) + ":");
					System.out.print(new String(kv.getQualifier()) + " ");
					// System.out.print(kv.getTimestamp() + " ");
					logger.debug(new String(kv.getValue()));
					String s = new String(kv.getValue());
					Double avgDouble = Double.valueOf(s);
					avgs.put(new String(kv.getQualifier()), avgDouble);
				}
			}

			// scanner.close();
			// table.close();

		} catch (IOException e) {
			e.printStackTrace();
		}

		return avgs;
	}

	private static FilterList applyFilters(String rowKey,
			AnalyticAvgInput input, String dateFromStr, String dateToStr) {
		// Create a list of filters to be applied
		List<Filter> filters = new ArrayList<Filter>();
		// filtering RowKey
		Filter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL,
				new RegexStringComparator(rowKey));
		filters.add(rowFilter);

		// filtering FamilyColumn
		Filter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL,
				new RegexStringComparator(input.getTimeScale().trim()));
		filters.add(familyFilter);

		// filtering qualifier "from" time range
		Filter qualifierFilterFrom = new QualifierFilter(
				CompareOp.GREATER_OR_EQUAL, new BinaryComparator(
						dateFromStr.getBytes()));
		filters.add(qualifierFilterFrom);

		// filtering qualifier "to" time range
		Filter qualifierFilterTo = new QualifierFilter(CompareOp.LESS_OR_EQUAL,
				new BinaryComparator(dateToStr.getBytes()));
		filters.add(qualifierFilterTo);

		FilterList filterList = new FilterList(
				FilterList.Operator.MUST_PASS_ALL, filters);
		return filterList;
	}

	public static void main(String[] agrs) {
		try {
			String csvFile = "/Volumes/Multimedia/shared/data/ucl-data-sample4.csv";
			BufferedReader br = null;
			String line = "";
			try {
				br = new BufferedReader(new FileReader(csvFile));
				while ((line = br.readLine()) != null) {
					String sourceid = "";
					String deviceid = "";
					String location = "";
					String devicetype = "";
					String group = "";
					String timestamp = "";
					String usage = "";
					String research = "";

					String[] svals = line.split(",");
					research = svals[0];
					System.out.println("research id::" + research);
					group = svals[1];
					System.out.println("group id::" + group);
					sourceid = svals[2];
					System.out.println("source id::" + sourceid);
					devicetype = svals[3];
					System.out.println("device type::" + devicetype);
					deviceid = svals[4];
					System.out.println("device id::" + deviceid);
					location = svals[5];
					System.out.println("location::" + location);
					usage = svals[6];
					System.out.println("usage::" + usage);
					timestamp = svals[7];
					System.out.println("timestamp::" + timestamp);

					String pkey = research + "/" + group + "/" + sourceid + "/"
							+ deviceid;

					ReadingDetails reading1 = new ReadingDetails(research,
							group, sourceid, deviceid, devicetype, location,
							timestamp, usage);
					String s = MSRResource.buildDetails(reading1);
					Date rTime = DateUtil.getDate(reading1.getRecordTime());
					String rTimeStr = DateUtil.makeFixedLength(String
							.valueOf(rTime.getTime()));
					MSRDao.addRecord(pkey.trim(), "details", rTimeStr, s.trim());
				}

			} catch (FileNotFoundException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			} finally {
				if (br != null) {
					try {
						br.close();
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			}
			System.out.println("Done");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

};
